package logs

import (
	"context"
	"encoding/json"
	gorillaWs "github.com/gorilla/websocket"
	"github.com/kataras/iris/v12/websocket"
	"github.com/kataras/neffos"
	"github.com/kataras/neffos/gorilla"
	"github.com/olivere/elastic/v7"
	streamCtx "gluenet/internal/glue-datasyncer/context"
	apis2 "gluenet/pkg/apis"
	"gluenet/pkg/ops"
	"net/http"
	"reflect"
	"strconv"
	"strings"
	"time"
)

const EsLogMapping = `
{
  "mappings": {
    "properties": {
      "agent_guid": {
        "type": "text"
      },
      "application_guid": {
        "type": "text"
      },
      "container": {
        "type": "text"
      },
      "level": {
        "type": "keyword"
      },
      "message": {
        "type": "text"
      },
      "object": {
        "type": "text"
      },
      "organize": {
        "type": "text"
      },
      "platform": {
        "type": "text"
      },
      "stream": {
        "type": "text"
      },
      "structured": {
        "type": "text"
      },
      "timestamp": {
        "type": "long"
      }
    }
  }
}`

// EsLog structure of EsLog
type EsLog struct {
	AgentGuid       string `json:"agent_guid"`
	ApplicationGuid string `json:"application_guid"`
	Container       string `json:"container"`
	Level           string `json:"level"`
	Message         string `json:"message"`
	Object          string `json:"object"`
	Organize        string `json:"organize"`
	Platform        string `json:"platform"`
	Stream          string `json:"stream"`
	Structured      string `json:"structured"`
	Timestamp       int    `json:"timestamp"`
}

// Logs Push logs to ElasticSearch
func Logs(ctx *streamCtx.StreamContext) {
	//Get client index
	client := ctx.DB.Es.Client
	c := context.Background()
	//Check if es index exists, if not then create it
	exists, err := client.IndexExists(ops.EsLog).Do(c)
	if err != nil {
		ctx.Logger.Errorf("check es index failed " + err.Error())
	}
	if !exists {
		_, err := client.CreateIndex(ops.EsLog).BodyString(EsLogMapping).Do(c)
		if err != nil {
			ctx.Logger.Errorf("create es index failed ", err.Error())
		}
	}
	//Push logs to es
	_, err = client.Index().Index(ops.EsLog).BodyString(string(ctx.RpcContext.Data.Data)).Do(c)
	if err != nil {
		ctx.Logger.Errorf("push logs to es failed ", err.Error())
	}
	ctx.Logger.Debugf("push log %s", ctx.RpcContext.Data.Data)
}

// LogsCount 日志统计
func LogsCount(c *streamCtx.StreamContext) {
	//Get es client index
	client := c.DB.Es.Client
	//Get url parameter of request
	startTime := c.WebContext.URLParam("start")
	guid := c.WebContext.URLParam("guid")
	container := c.WebContext.URLParam("container")
	endTime := c.WebContext.URLParam("end")
	bucket := c.WebContext.URLParam("bucket")

	//Check if instance exist
	//var ins *meta.Instance
	//if err := c.DB.InstanceGet(guid, &ins); err != nil {
	//	c.Logger.Errorf("query instance error :", err.Error())
	//	c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
	//	return
	//}

	//1.Query from db
	//Set criteria for the query
	boolQuery := elastic.NewBoolQuery()
	termQueryGUID := elastic.NewQueryStringQuery("application_guid:" + guid)
	termQueryContainer := elastic.NewQueryStringQuery("container:" + container)
	rangeQueryTimeStamp := elastic.NewRangeQuery("timestamp").Gte(startTime).Lte(endTime)
	boolQuery.Must(termQueryGUID, termQueryContainer, rangeQueryTimeStamp)
	//Set the aggregation with interval of timestamp and term of level
	//ParseDuration does not support the format like "1d"
	if strings.Contains(bucket, "d") {
		count, _ := strconv.Atoi(bucket[:strings.Index(bucket, "d")])
		hour := strconv.Itoa(count * 24)
		bucket = hour + "h"
	}
	interval, _ := time.ParseDuration(bucket)
	timeStamp := elastic.NewHistogramAggregation().Field("timestamp").Interval(interval.Seconds()).MinDocCount(0)
	histogram := elastic.NewTermsAggregation().Field("level").MinDocCount(0)
	timeStamp.SubAggregation("level", histogram)
	//Start query
	searchResult, err := client.Search().
		Index(ops.EsLog).
		Query(boolQuery).
		Aggregation("timestamp", timeStamp).
		Size(0).
		Do(context.Background())
	if err != nil {
		c.Logger.Errorf("query error :", err.Error())
		c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
		return
	}
	//2.Data organization
	aggResByTimeStamp, found := searchResult.Aggregations.Histogram("timestamp")
	if !found {
		c.Logger.Errorf("no aggregate results found with timestamp,error :", err.Error())
		c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
		return
	}
	var res []map[string]interface{}
	for _, bucket := range aggResByTimeStamp.Buckets {
		bucketValue := bucket.Key
		aggResByLevel, found := bucket.Aggregations.Terms("level")
		if !found {
			c.Logger.Errorf("no aggregate results found with level,error :", err.Error())
			c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
			return
		}
		for _, log := range aggResByLevel.Buckets {
			res = append(res, map[string]interface{}{
				"counts": log.DocCount,
				"label":  "logs",
				"start":  int(bucketValue),
				"stop":   int(bucketValue + interval.Seconds()),
				"type":   log.Key,
			})
		}
	}
	//3.Response
	c.WebContext.JSON(apis2.ResponseOK(res))
}

// LogsInfoList 日志信息
func LogsInfoList(c *streamCtx.StreamContext) {
	//Get es client index
	client := c.DB.Es.Client
	//Get url parameter of request
	startTime := c.WebContext.URLParam("start")
	guid := c.WebContext.URLParam("guid")
	container := c.WebContext.URLParam("container")
	endTime := c.WebContext.URLParam("end")

	////Check if instance exist
	//var ins *meta.Instance
	//if err := c.DB.InstanceGet(guid, &ins); err != nil {
	//	c.Logger.Errorf("query instance error :", err.Error())
	//	c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
	//	return
	//}

	//1.Query from db
	//Set criteria for the query
	boolQuery := elastic.NewBoolQuery()
	termQueryGUID := elastic.NewQueryStringQuery("application_guid:" + guid)
	termQueryContainer := elastic.NewQueryStringQuery("container:" + container)
	rangeQueryTimeStamp := elastic.NewRangeQuery("timestamp").Gte(startTime).Lte(endTime)
	boolQuery.Must(termQueryGUID, termQueryContainer, rangeQueryTimeStamp)
	//Start query
	searchResult, err := client.Search().
		Index(ops.EsLog).
		Query(boolQuery).
		Sort("timestamp", true).
		Do(context.Background())
	if err != nil {
		c.Logger.Errorf("query error :", err.Error())
		c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
		return
	}
	//2.Data organization
	var res []map[string]interface{}
	if searchResult.TotalHits() > 0 {
		var esLog EsLog
		for _, item := range searchResult.Each(reflect.TypeOf(esLog)) {
			if t, ok := item.(EsLog); ok {
				res = append(res, map[string]interface{}{
					"label":   "logs",
					"message": t.Message,
					"time":    t.Timestamp,
					"type":    t.Level,
				})
			}
		}
	}
	//3.Response
	c.WebContext.JSON(apis2.ResponseOK(res))
}

// LogsSocket 日志Socket
func LogsSocket(c *streamCtx.StreamContext) {
	//Get es client index
	client := c.DB.Es.Client
	guid := c.WebContext.URLParam("guid")
	container := c.WebContext.URLParam("container")

	////Check if instance exist
	//var ins *meta.Instance
	//if err := c.DB.InstanceGet(guid, &ins); err != nil {
	//	c.Logger.Errorf("query instance error :", err.Error())
	//	c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
	//	return
	//}

	//1.Create websocket server
	ctx, cancel := context.WithCancel(c.Context)
	var websocketServer = websocket.New(gorilla.Upgrader(gorillaWs.Upgrader{
		CheckOrigin: func(*http.Request) bool { return true },
	}), websocket.Events{
		websocket.OnNativeMessage: func(conn *neffos.NSConn, message neffos.Message) error { return nil },
	})
	websocketServer.OnConnect = func(n *websocket.Conn) error {
		c.Logger.Debugf("OnConnect from server %v", n.ID())
		tk := time.NewTicker(time.Second * 5)
		for {
			select {
			case <-tk.C:
				endTime := time.Now().Unix()
				startTime := endTime - 4
				//2.Query from db
				//Set criteria for the query
				boolQuery := elastic.NewBoolQuery()
				termQueryGUID := elastic.NewQueryStringQuery("application_guid:" + guid)
				termQueryContainer := elastic.NewQueryStringQuery("container:" + container)
				rangeQueryTimeStamp := elastic.NewRangeQuery("timestamp").Gte(startTime).Lte(endTime)
				boolQuery.Must(termQueryGUID, termQueryContainer, rangeQueryTimeStamp)
				//Start query
				searchResult, err := client.Search().
					Index(ops.EsLog).
					Query(boolQuery).
					Sort("timestamp", true).
					Do(context.Background())
				if err != nil {
					c.Logger.Errorf("query error :", err.Error())
					c.WebContext.JSON(apis2.ResponseReqErr(err, nil))
				}
				//3.Data organization
				var res []map[string]interface{}
				if searchResult.TotalHits() > 0 {
					var esLog EsLog
					for _, item := range searchResult.Each(reflect.TypeOf(esLog)) {
						if t, ok := item.(EsLog); ok {
							res = append(res, map[string]interface{}{
								"label":   "logs",
								"message": t.Message,
								"time":    t.Timestamp,
								"type":    t.Level,
							})
						}
					}
				}
				//4.Send data to client
				bts, _ := json.Marshal(res)
				n.Write(neffos.Message{
					Body:     bts,
					IsNative: true,
				})
			case <-ctx.Done():
				c.Logger.Debugf("push down %v", n.ID())
				return nil
			}
		}
	}
	websocketServer.OnDisconnect = func(n *websocket.Conn) {
		cancel()
		c.Logger.Debugf("Disconnected from server", n.ID())
	}
	websocketServer.OnUpgradeError = func(err error) {
		cancel()
		c.Logger.Debugf("OnUpgradeError from server %v", err)
	}
	websocket.Handler(websocketServer)(c.WebContext)
}
